--
-- The test measures run time of space full scan.
--
-- Output format (console):
-- <test-case> <rows-per-second>
--
-- NOTE: The test requires a C module. Set the BUILDDIR environment variable to
-- the tarantool build directory if using out-of-source build.
--

require('compat').box_error_serialize_verbose = 'new'

local clock = require('clock')
local fiber = require('fiber')
local fio = require('fio')
local log = require('log')
local tarantool = require('tarantool')
local benchmark = require('benchmark')

local USAGE = [[
   engine <string, 'memtx'>         - space engine to use for the test
   column_count <number, 100>       - number of columns in the test space
                                      (must be 2 if field_type is string)
   row_count <number, 1000000>      - number of rows in the test space
   field_type <string, 'unsigned'>  - type of the fields in the test space
                                      (supported types: unsigned, string)
   str_len_min <number, 1>          - minimum length (if field_type is string)
   str_len_max <number, 100>        - maximum length (if field_type is string)
   fill_ratio <number, 1.0>         - number of non-NULL values / row_count
   layout <string, 'plain'>         - column layout option of the index
   use_read_view <boolean, false>   - use a read view
   use_arrow_api <boolean, false>   - use the arrow stream API
   scan_backwards <boolean, false>  - scan backwards
   no_reuse_data <boolean, false>   - do not reuse data generated by last run
   minimum_run_time <number, 5>     - minimal time to run distinct perf test
                                      in seconds

 Being run without options, this benchmark measures the run time of a full scan
 from the space.
]]

local params = benchmark.argparse(arg, {
    {'engine', 'string'},
    {'column_count', 'number'},
    {'row_count', 'number'},
    {'field_type', 'string'},
    {'str_len_min', 'number'},
    {'str_len_max', 'number'},
    {'fill_ratio', 'number'},
    {'layout', 'string'},
    {'use_read_view', 'boolean'},
    {'use_arrow_api', 'boolean'},
    {'scan_backwards', 'boolean'},
    {'no_reuse_data', 'boolean'},
    {'minimum_run_time', 'number'},
}, USAGE)

params.engine = params.engine or 'memtx'
params.column_count = params.column_count or 100
params.row_count = params.row_count or 1000 * 1000
params.field_type = params.field_type or 'unsigned'
params.str_len_min = params.str_len_min or 1
params.str_len_max = params.str_len_max or 100
params.fill_ratio = params.fill_ratio or 1
params.layout = params.layout or 'plain'
params.use_read_view = params.use_read_view or false
params.use_arrow_api = params.use_arrow_api or false
params.scan_backwards = params.scan_backwards or false
params.no_reuse_data = params.no_reuse_data or false
params.minimum_run_time = params.minimum_run_time or 5

local bench = benchmark.new(params)

local BUILDDIR = fio.abspath(fio.pathjoin(os.getenv('BUILDDIR') or '.'))
local MODULEPATH = fio.pathjoin(BUILDDIR, 'perf', 'lua',
                                '?.' .. tarantool.build.mod_format)

package.cpath = MODULEPATH .. ';' .. package.cpath

local function die(fmt, ...)
    local msg = fmt:format(...)
    if log.cfg.log ~= nil then
        log.error(msg)
    end
    io.stderr:write(msg .. '\n')
    os.exit(1)
end

local function do_or_die(func, ...)
    local ok, err = pcall(func, ...)
    if not ok then
        die('%s', err)
    end
end

if params.field_type ~= 'unsigned' and params.field_type ~= 'string' then
    die("Unsupported field type")
end

if params.field_type == 'string' then
    if params.column_count ~= 2 then
        die('Field type "string" requires exactly 2 columns')
    end
    if params.str_len_min > params.str_len_max then
        die('Minimum length cannot be greater than maximum length')
    end
end

if params.fill_ratio > 1 then
    die("Fill ratio must be in the range from 0 to 1")
end

local has_column_scan, test_module = pcall(require, 'column_scan_module')
if not has_column_scan then
    die('Lua module "column_scan_module" is not found')
end
local test_funcs = {}

for _, func_name in ipairs({'sum', 'str'}) do
    local full_func_name
    if params.use_arrow_api then
        full_func_name = func_name .. '_arrow'
    else
        full_func_name = func_name .. '_iterator'
    end
    if params.use_read_view then
        full_func_name = full_func_name .. '_rv'
    end
    local f = test_module[full_func_name]
    if f == nil then
        error('The specified test mode is not supported by this build')
    end
    test_funcs[func_name] = f
end

local FIELD_TYPE = params.field_type
if FIELD_TYPE == 'string' then
    FIELD_TYPE = string.format(
        '%s:%d-%d', FIELD_TYPE, params.str_len_min, params.str_len_max)
end
local WORK_DIR = string.format(
    'column_scan,engine=%s,column_count=%d,row_count=%d,field_type=%s,' ..
    'fill_ratio=%f,layout=%s',
    params.engine, params.column_count, params.row_count, FIELD_TYPE,
    params.fill_ratio, params.layout)

if params.no_reuse_data then
    fio.rmtree(WORK_DIR)
end

fio.mkdir(WORK_DIR)

box.cfg({
    work_dir = WORK_DIR,
    log = 'tarantool.log',
    memtx_memory = 4 * 1024 * 1024 * 1024,
    checkpoint_count = 1,
})

local gen_arrow = test_module.gen_arrow

local function gen_insert(space_id, column_count, row_count, field_type,
                          str_len_min, str_len_max, fill_ratio)
    local s = box.space[space_id]
    local tuple = {}
    local pct_complete = 0
    log.info('Generating the test data set...')
    box.begin()
    for i = 1, row_count do
        if fill_ratio <= math.random() then
            tuple[1] = i
            for j = 2, column_count do
                tuple[j] = box.NULL
            end
        else
            if field_type == 'unsigned' then
                for j = 1, column_count do
                    if j % 2 == 1 then
                        tuple[j] = i
                    else
                        tuple[j] = row_count - i + 1
                    end
                end
            elseif field_type == 'string' then
                local char = string.char(97 + ((i - 1) % 26))
                local len = math.random(str_len_min, str_len_max)
                tuple = {i, string.rep(char, len)}
            else
                error("Unsupported field type")
            end
        end
        s:insert(tuple)
        if i % 1000 == 0 then
            box.commit()
            local pct = math.floor(100 * i / row_count)
            if pct ~= pct_complete then
                log.info('%d%% complete', pct)
                pct_complete = pct
            end
            box.begin()
        end
    end
    box.commit()
end

do_or_die(box.once, 'init', function()
    log.info('Creating the test space...')
    local format = {}
    table.insert(format, {'field_1', 'unsigned'})
    for i = 2, params.column_count do
        table.insert(format, {
            'field_' .. i, params.field_type, is_nullable = true,
        })
    end
    local s = box.schema.space.create('test', {
        engine = params.engine,
        field_count = #format,
        format = format,
    })
    if params.layout == 'plain' then
        s:create_index('pk')
    else
        s:create_index('pk', {layout = params.layout})
    end
    local gen
    if params.engine == 'memcs' or params.engine == 'quiver' then
        gen = gen_arrow
    else
        gen = gen_insert
    end
    gen(s.id, params.column_count, params.row_count, params.field_type,
        params.str_len_min, params.str_len_max, params.fill_ratio)
    log.info('Writing a snapshot...')
    box.snapshot()
end)

local TESTS = {
    {
        name = 'sum_first',
        type = 'unsigned',
        func = function()
            return test_funcs.sum(box.space.test.id, 0, 0,
                                  params.scan_backwards)
        end,
    },
    {
        name = 'sum_last',
        type = 'unsigned',
        func = function()
            return test_funcs.sum(box.space.test.id, 0,
                                  params.column_count - 1,
                                  params.scan_backwards)
        end,
    },
    {
        name = 'string_notouch',
        type = 'string',
        func = function()
            return test_funcs.str(box.space.test.id, 0, 1, false, false,
                                  params.scan_backwards)
        end,
    },
    {
        name = 'string_touch',
        type = 'string',
        func = function()
            return test_funcs.str(box.space.test.id, 0, 1, false, true,
                                  params.scan_backwards)
        end,
    },
    {
        name = 'string_view_notouch',
        type = 'string',
        func = function()
            return test_funcs.str(box.space.test.id, 0, 1, true, false,
                                  params.scan_backwards)
        end,
    },
    {
        name = 'string_view_touch',
        type = 'string',
        func = function()
            return test_funcs.str(box.space.test.id, 0, 1, true, true,
                                  params.scan_backwards)
        end,
    },
}

local function run_test(test)
    local func = test.func
    local supported = func() -- warmup
    if not supported then
        return
    end
    local real_time_start = clock.time()
    local cpu_time_start = clock.proc()
    local delta_real
    local count = 0
    repeat
        supported = func()
        if not supported then
            die('Unexpected test skipping')
        end
        delta_real = clock.time() - real_time_start
        count = count + 1
    until delta_real > params.minimum_run_time
    local delta_cpu = clock.proc() - cpu_time_start
    bench:add_result(test.name, {
        real_time = delta_real,
        cpu_time = delta_cpu,
        items = params.row_count * count,
    })
end

fiber.set_max_slice(9000)

local seed = os.time()
log.info('Random seed: ' .. seed)
math.randomseed(seed)

for _, test in ipairs(TESTS) do
    if test.type == params.field_type then
        log.info('Running test %s...', test.name)
        do_or_die(run_test, test)
    end
end

bench:dump_results()

os.exit(0)
